package org.kf

import java.util.Properties

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}

/**
 * description ：将读取到的数据写入 kafka 中
 * author      ：剧情再美终是戏 
 * mail        : 13286520398@163.com
 * date        ：Created in 2020/2/25 18:30
 * modified By ：
 * version:    : 1.0
 */
object KafkaWrite {
  def main(args: Array[String]): Unit = {
    // 读取文件的数据
    val source = io.Source.fromFile("F:\\Word\\idea\\bigdata\\Flink0830s\\src\\main\\resources\\UserBehavior.csv")

    // 写入 kafka
    val properties = new Properties()
    properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop101:9092")
    properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")
    properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")
    val produce = new KafkaProducer[String, String](properties)
    for (line <- source.getLines()) {
      produce.send(new ProducerRecord("t01", line))
    }
    produce.close()
  }
}
